package OrderState_jiankong;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

/**
 * @author Spring_Hu
 * @date 2021/10/13 18:03
 */
public class OrderState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<OrderEvent> ods = env.readTextFile("test/OrderLog.csv").map(new MapFunction<String, OrderEvent>() {
            @Override
            public OrderEvent map(String value) throws Exception {
                String[] datas = value.split(",");
                return new OrderEvent(Long.parseLong(datas[0]), datas[1]
                        , datas[2], Long.parseLong(datas[3]));
            }
        });



        SingleOutputStreamOperator<TxEvent> txs = env.readTextFile("test/ReceiptLog.csv").map(new MapFunction<String, TxEvent>() {
            @Override
            public TxEvent map(String value) throws Exception {
                String[] datas = value.split(",");
                return new TxEvent(datas[0], datas[1]
                        , Long.parseLong(datas[2]));
            }
        });

        ConnectedStreams<OrderEvent, TxEvent> connect = ods.connect(txs);

        ConnectedStreams<OrderEvent, TxEvent> keybySource = connect.keyBy("txId", "txId");

        keybySource.process(new CoProcessFunction<OrderEvent, TxEvent, String>() {
            //存放(txId,orderEvent)
            HashMap<String,OrderEvent> orderMap=new HashMap<>();
            //存放（txId,TxEvent）
            HashMap<String,TxEvent> txMap=new HashMap<>();
            @Override
            public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {

                    if(txMap.containsKey(value.getTxId())){
                        out.collect("订单: " + value + " 对账成功");
                        txMap.remove(value.getTxId());
                    }else{
                        orderMap.put(value.getTxId(),value);
                    }
            }

            @Override
            public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                    if(orderMap.containsKey(value.getTxId())){
                        out.collect("订单: " + value + " 对账成功");
                        orderMap.remove(value.getTxId());
                    }else {
                        txMap.put(value.getTxId(),value);
                    }

            }
        }).print();

        env.execute();


    }

}
